package com.lzcu.maoyan.mapreduce.mapreduce.reduce;

import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieTypeWritable;
import com.lzcu.maoyan.mapreduce.service.impl.PhoenixMapperImpl;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author WangLiang
 * @date `2024/4/24`
 */
public class MovieTypeReduce extends Reducer<Text, Text, NullWritable, MovieTypeWritable> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        int count = 0;
        long maxMovieTotal = 0;
        String maxMovieName = "";
        for (Text value : values) {
            String movieName = value.toString();
            long movieBoxOffice;
            PhoenixMapperImpl phoenixMapper = new PhoenixMapperImpl();
            try {
                movieBoxOffice = phoenixMapper.getMovieBoxOffice(movieName);
            } catch (Exception e){
                movieBoxOffice = 0;
                e.printStackTrace();
            }
            if (maxMovieTotal < movieBoxOffice){
                maxMovieTotal = movieBoxOffice;
                maxMovieName = movieName;
            }
            count += 1;
        }
        MovieTypeWritable movieTypeWritable = new MovieTypeWritable();
        movieTypeWritable.setMovieType(key.toString());
        movieTypeWritable.setCount(count);
        movieTypeWritable.setMovieName(maxMovieName);
        movieTypeWritable.setTotalSessions(maxMovieTotal);

        context.write(NullWritable.get(), movieTypeWritable);
    }
}
